﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using RabbitMQ.Client;

namespace ConsoleApp.Infrastructure.Rmq
{
    public class RmqConnectPool
    {
        private List<RmqConnect> _cList = new List<RmqConnect>();

        private object _lock = new object();

        private string _connect;

        private int _poolSize;

        private int _checkSleepSecs;

        private ConnectionFactory _factory;

        private Random _rd;

        private ReaderWriterLockSlim _readWriteLock;

        public ConnectionFactory GetInnerFactoryObject()
        {
            return _factory;
        }

        public RmqConnectPool(string uri, int poolSize, int checkSleepSecs = 3)
        {
            _connect = uri;
            _poolSize = poolSize;
            _checkSleepSecs = checkSleepSecs;
            _rd = new Random();
            _readWriteLock = new ReaderWriterLockSlim();
            _factory = new ConnectionFactory
            {
                Uri = new Uri(uri),
                AutomaticRecoveryEnabled = false,
                RequestedHeartbeat = 3,
                RequestedConnectionTimeout = 10000,
                SocketWriteTimeout = 300000,
                SocketReadTimeout = 300000,
                NetworkRecoveryInterval = TimeSpan.FromSeconds(3.0)
            };
            new Thread((ThreadStart)delegate
            {
                try
                {
                    ConnectCheck();
                }
                catch (Exception ex2)
                {
                    RmqLogHelper.WriteError("数据库心跳检测异常：" + ex2.ToString());
                }
            }).Start();
            try
            {
                using IConnection connection = _factory.CreateConnection();
                using IModel model = connection.CreateModel();
                model.ExchangeDeclare(RmqBasicInfoConfig.BaseExchange, "direct", durable: true);
            }
            catch (Exception ex)
            {
                RmqLogHelper.WriteError("基础exchange创建出错：" + ex.ToString());
            }
        }

        public RmqConnect GetConnect()
        {
            _readWriteLock.EnterUpgradeableReadLock();
            RmqConnect result = null;
            try
            {
                if (_cList.Where((p) => p.GetRealConnectInfo().IsOpen).Count() < _poolSize)
                {
                    try
                    {
                        _readWriteLock.EnterWriteLock();
                        RmqConnect item = new RmqConnect(_factory.CreateConnection());
                        RmqLogHelper.WriteInfo("rmq 生成链接,当前连接数:" + _cList.Count);
                        _cList.Add(item);
                        _cList.RemoveAll((p) => !p.GetRealConnectInfo().IsOpen);
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine(ex.ToString());
                    }
                    finally
                    {
                        _readWriteLock.ExitWriteLock();
                    }
                }

                return (from p in _cList
                        where p.GetRealConnectInfo().IsOpen
                        orderby Guid.NewGuid() descending
                        select p).FirstOrDefault();
            }
            catch (Exception ex2)
            {
                Console.WriteLine(ex2.ToString());
                return result;
            }
            finally
            {
                _readWriteLock.ExitUpgradeableReadLock();
            }
        }

        private void ConnectCheck()
        {
            while (true)
            {
                Thread.Sleep(1000 * _checkSleepSecs);
                _readWriteLock.EnterReadLock();
                List<RmqConnect> list = new List<RmqConnect>();
                try
                {
                    for (int i = 0; i < _cList.Count; i++)
                    {
                        RmqConnect rmqConnect = _cList[i];
                        try
                        {
                            using IModel model = rmqConnect.GetChannel();
                            model.ExchangeDeclare("__HeartChcekRmqPool", "fanout");
                        }
                        catch (Exception ex)
                        {
                            RmqLogHelper.WriteInfo("rmq connectcheck 异常，加入到异常连接池:" + ex.ToString());
                            try
                            {
                                IConnection realConnectInfo = rmqConnect.GetRealConnectInfo();
                                if (realConnectInfo != null && realConnectInfo.IsOpen)
                                {
                                    realConnectInfo.Close();
                                }
                            }
                            catch (Exception)
                            {
                            }

                            list.Add(rmqConnect);
                        }
                    }
                }
                finally
                {
                    _readWriteLock.ExitReadLock();
                }

                if (list.Count <= 0)
                {
                    continue;
                }

                _readWriteLock.EnterWriteLock();
                try
                {
                    foreach (RmqConnect item in list)
                    {
                        RmqLogHelper.WriteError("rmq connectcheck 异常，移除链接:" + _cList.Remove(item));
                    }
                }
                finally
                {
                    _readWriteLock.ExitWriteLock();
                }
            }
        }
    }
}
